package net.bwie.realtime.warehouse.DWS;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.warehouse.DWS
 * @Author: zhangleqing
 * @CreateTime: 2025-09-02  09:42
 * @Description: TODO 退中检测 dws_refund_recovery （退款挽回表）
 * 指标：pay_time（订单支付时间，实时每一天） product_id （商品id） product_category（商品品类）
 * processing_refund_amount（处理中退款金额） refund_recover_ rate（退款挽回率）
 * @Version: 1.0
 */
public class DwsRefundRecovery {
    private static TableEnvironment tableEnv;

    public static void main(String[] args) {
        // 1 开启上下文
        TableEnvironment tableEnv = getTableEnv();

        // 2 读取数据
        readTable(tableEnv);

        // 3 数据计算
        Table resultTable = handle(tableEnv);

        resultTable.execute().print();

        // 4 映射表创建

        // 5 输出数据
    }


/*
-- 步骤1：计算前1年各品类的挽回率（用于后续计算预计可挽回金额）
WITH category_recover_rate AS (
    SELECT
        product_category,
        -- 处理订单总金额为0的特殊情况，避免除法错误
        CASE
            WHEN SUM(order_amount) = 0 THEN 0
            ELSE ROUND(
                (SUM(order_amount - refund_amount) / SUM(order_amount)) * 100,
                2  -- 保留2位小数
            )
        END AS category_rate  -- 品类挽回率（百分比）
    FROM dwd_Step_back
    -- 时间范围：前1年（不含当天，避免数据不完整）
    WHERE pay_time BETWEEN
        DATEADD(YEAR, -1, CURRENT_DATE)  -- 修正DATEADD参数顺序：单位、偏移量、基准日期
        AND CURRENT_DATE - INTERVAL '1' DAY  -- 昨天
    GROUP BY product_category
),
-- 步骤2：计算当日处理中退款金额（按商品ID分组）
daily_processing_refund AS (
    SELECT
        product_id,
        product_category,
        SUM(refund_amount) AS processing_refund_amount
    FROM dwd_Step_back
    -- 筛选条件1：当日发起的退款（按pay_time日期判断）
    WHERE DATE_FORMAT(pay_time, 'yyyy-MM-dd') = DATE_FORMAT(CURRENT_DATE, 'yyyy-MM-dd')
    -- 筛选条件2：退款状态未完结（根据实际业务状态调整）
    AND refund_status IN ('pending', 'waiting_for_return', 'processing')  -- 未完结状态示例
    GROUP BY product_id, product_category
)
-- 最终结果：关联计算两个指标
SELECT
    dpr.product_id,
    dpr.product_category,
    dpr.processing_refund_amount,  -- 处理中退款金额
    -- 预计可挽回金额 = 处理中退款金额 × 品类挽回率（百分比转小数），用COALESCE处理NULL为0
    ROUND(
        dpr.processing_refund_amount * (COALESCE(crr.category_rate, 0) / 100),
        2  -- 保留2位小数
    ) AS expected_recoverable_amount  -- 修正字段名，明确为“预计可挽回金额”
FROM daily_processing_refund dpr
-- 左连接确保即使品类无历史数据也能保留商品记录（此时挽回率为0）
LEFT JOIN category_recover_rate crr
    ON dpr.product_category = crr.product_category;
 */

    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                ""
        );
        return table;
    }

    private static void readTable(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "CREATE TABLE dwd_Step_back (\n" +
                        "    product_category STRING,\n" +
                        "    product_id STRING,\n" +
                        "    refund_id STRING,\n" +
                        "    refund_status STRING,\n" +
                        "    refund_amount DOUBLE,\n" +
                        "    order_id STRING,\n" +
                        "    order_amount DOUBLE,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        // 定义水位线：基于pay_time，允许1天的延迟
                        "    WATERMARK FOR pay_time AS pay_time - INTERVAL '1' DAY\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dwd_Step_back',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'format' = 'json',\n" +
                        // 读取模式配置：从最早位置开始消费
                        "    'scan.startup.mode' = 'earliest-offset'\n" +
                        // 也可以根据需要改为从最新位置消费
                        // 'scan.startup.mode' = 'latest-offset'\n" +
                        ")"
        );
    }


    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态TTL设置为25小时（3600*25=90000秒），覆盖24小时的关联区间
        configuration.setString("table.exec.state.ttl", "90000 s");
//        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }
}
